package com.steve.bigdata.flinkmonitor.assist;

import com.steve.bigdata.flinkmonitor.model.WindowResult;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @Author: STEVE
 * @Description: 窗口结果构造器
 * 功能：将聚合结果(Tuple2<sum, count>)与窗口元数据组合成完整的WindowResult
 * @since: 2025-07-10
 */
public class WindowResultBuilder implements WindowFunction<
        Tuple2<Double, Long>,   // 输入：聚合结果（总和，计数）
        WindowResult,           // 输出：完整窗口结果
        String,                 // 键类型（userId）
        TimeWindow> {           // 窗口类型

    /**
     * 构建最终窗口结果
     *
     * @param userId 窗口键（userId）
     * @param window 窗口元数据
     * @param input  聚合结果迭代器（实际只有一个元素）
     * @param out    结果收集器
     */
    @Override
    public void apply(String userId, TimeWindow window, Iterable<Tuple2<Double, Long>> input, Collector<WindowResult> out) {
        // 获取聚合结果（注意：Iterable实际只有一个元素）
        Tuple2<Double, Long> aggregationResult = input.iterator().next();
        // 构造完整窗口结果
        WindowResult result = new WindowResult(
                userId,                 // 用户ID
                window.getStart(),      // 窗口开始时间（毫秒）
                window.getEnd(),        // 窗口结束时间（毫秒）
                aggregationResult.f0,   // 数值总和
                aggregationResult.f1    // 事件计数
        );
        // 输出结果
        out.collect(result);
        /* 设计说明：
         * 1. 此处的Iterable通常只包含一个元素，因为每个窗口对应一个聚合结果
         * 2. 对于会话窗口等可能合并的情况，需要遍历所有元素合并结果
         */
    }

}
